package com.googlecode.protoserv;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.ProtocolException;
import java.net.URL;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import com.google.protobuf.ExtensionRegistry;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.Message;
import com.google.protobuf.Message.Builder;
import com.googlecode.protoserv.Protocol.RequestProto;
import com.googlecode.protoserv.Protocol.ResponseProto;
import com.googlecode.protoserv.Protocol.RpcStatus;
import com.googlecode.protoserv.Transport.Encoding;

class HttpTransport extends RpcTransport implements HttpConstants {

  private static final String HTTP_ACCEPTED_ENCODINGS = Encoding.GZIP
      .toString();

  private URL url;
  private String authentication;

  public HttpTransport(URL url) {
    this.url = url;
  }

  public void setAuthentication(String type, String authentication) {
    this.authentication = type + " " + authentication;
  }

  private HttpURLConnection createConnection(Encoding reqEncoding, int reqSize)
      throws IOException {
    HttpURLConnection conn = (HttpURLConnection) this.url.openConnection();
    try {
      conn.setRequestMethod(POST_METHOD);
    } catch (ProtocolException e) {
      e.printStackTrace();
    }
    conn.setRequestProperty(CONNECTION, KEEP_ALIVE);
    conn.setRequestProperty(USER_AGENT, PROTOSERV_USER_AGENT_PREFIX + "/j2se");
    if (this.authentication != null) {
      conn.setRequestProperty(AUTHORIZATION, this.authentication);
    }
    conn.setRequestProperty(ACCEPT, MIME_PROTOBUF);
    conn.setRequestProperty(ACCEPT_ENCODING, HTTP_ACCEPTED_ENCODINGS);
    conn.setRequestProperty(CONTENT_TYPE, MIME_PROTOBUF);
    if (!reqEncoding.equals(Encoding.IDENTITY)) {
      conn.setRequestProperty(CONTENT_ENCODING, reqEncoding.name()
          .toLowerCase());
    }
    conn.setRequestProperty(CONTENT_LENGTH, Integer.toString(reqSize));
    conn.setDoOutput(true);
    conn.setDoInput(true);
    return conn;
  }

  private void consumeStream(InputStream stream) {
    byte[] buffer = new byte[32];
    try {
      while (stream.read(buffer, 0, buffer.length) > 0) {}
      stream.close();
    } catch (IOException e) {}
  }

  private static void encode(Encoding encoding, Message msg, ByteStream stream)
      throws IOException {
    int size = msg.getSerializedSize();
    stream.encsureCapacity(size);
    switch (encoding.getNumber()) {
      case Encoding.IDENTITY_VALUE:
        msg.writeTo(stream);
        return;
      case Encoding.GZIP_VALUE:
        GZIPOutputStream gzipOut = new GZIPOutputStream(stream);
        msg.writeTo(gzipOut);
        gzipOut.finish();
        gzipOut.close();
        return;
      default:
        throw new IOException("Unsupported encoding.");
    }
  }

  private static void decode(Encoding encoding, InputStream in,
      Builder message, ExtensionRegistry registry) throws IOException {
    switch (encoding.getNumber()) {
      case Encoding.IDENTITY_VALUE:
        message.mergeFrom(in, registry);
        return;
      case Encoding.GZIP_VALUE:
        GZIPInputStream gzipIn = new GZIPInputStream(in);
        message.mergeFrom(gzipIn, registry);
        gzipIn.close();
        return;
      default:
        throw new IOException("Unsupported encoding.");
    }
  }

  @Override
  @SuppressWarnings("rawtypes")
  public void send(Rpc rpc, RequestProto.Builder request) throws IOException {
    HttpURLConnection conn = null;
    OutputStream netOutput = null;
    InputStream netInput = null;
    try {
      // Send request.
      ByteStream reqStream = new ByteStream();
      encode(rpc.getTransportEncoding(), request.build(), reqStream);
      conn = createConnection(rpc.getTransportEncoding(), reqStream.size());
      netOutput = conn.getOutputStream();
      reqStream.writeTo(netOutput);
      netOutput.flush();
      netOutput.close();

      // Receive response.
      if ((conn.getResponseCode() / 100) != 2) {
        throw new IOException(conn.getResponseMessage());
      }
      String contentType = conn.getContentType();
      if (contentType == null || !contentType.equals(MIME_PROTOBUF)) {
        rpc.setResponse(RpcStatus.RPC_ERROR, "Invalid content type: "
            + contentType);
        return;
      }
      // Decode response.
      String encodingStr = conn.getContentEncoding();
      if (encodingStr == null || encodingStr.length() == 0) {
        encodingStr = Encoding.IDENTITY.name();
      }
      Encoding encoding = Encoding.valueOf(encodingStr.toUpperCase());
      if (encoding == null) {
        rpc.setResponse(RpcStatus.RPC_ERROR,
            "Invalid encoding: '" + conn.getContentEncoding() + "'");
        return;
      }
      try {
        ExtensionRegistry registry = ExtensionRegistry.newInstance();
        registry.add(rpc.getReponseExtension());
        ResponseProto.Builder response = ResponseProto.newBuilder();
        decode(encoding, conn.getInputStream(), response, registry);
        rpc.setResponse(response.build());
      } catch (InvalidProtocolBufferException e) {
        rpc.setResponse(RpcStatus.RPC_ERROR,
            "Invalid content: " + e.getMessage());
      }
    } finally {
      consumeStream(netInput);
      consumeStream(conn.getErrorStream());
      if (netOutput != null) {
        netOutput.close();
      }
      if (netInput != null) {
        netInput.close();
      }
      if (conn != null) {
        conn.disconnect();
      }
    }
  }
}
